package com.behavioranalysis.flinkprogram.flink.ad;

import com.behavioranalysis.flinkprogram.conf.ConfigurationManager;
import com.behavioranalysis.flinkprogram.constant.Constants;
import com.behavioranalysis.flinkprogram.dao.output.AdClickTrendSink;
import com.behavioranalysis.flinkprogram.dao.output.AdRealTimeStatSink;
import com.behavioranalysis.flinkprogram.dao.output.BlackSink;
import com.behavioranalysis.flinkprogram.dao.output.DailyUserAdClickCountSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.*;

/**
 * 广告点击流量实时统计Flink作业
 */
public class AdClickRealTimeStatKafka2Flink2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = getEnv();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<Tuple6<Long, String, String, String, String, String>> adRealTimeLogTuple = env
                .addSource(getKafkaConsumer())
                .map(new SourceString2TupleMapFunction());
        //adRealTimeLogTuple.print();

        // 侧面输出流： clickCount >= 100 的用户userid
        final OutputTag<Long> blacklistUserid = new OutputTag<Long>("blacklistUserid") {
        };

        /**
         * 一： 设置滚动窗口，进行黑名单用户过滤
         *      窗口大小：两分钟，120秒
         *      数据库黑名单表按流来读，行不通
         *      按静态数据来读
         *
         *      TODO join 黑名单表  过滤
         */

        // 使用处理时间ProcessingTime，不需要分配时间戳和生成水印，因为处理时间是程序处理数据的当前时间，直接生成
        // 而使用事件时间，程序需要使用直接为数据定义事件时间并自己发出水印的源，或者程序必须在源之后注入Timestamp Assigner＆Watermark Generator。这些功能描述了如何访问事件时间戳，以及事件流呈现出何种程度的乱序。
        // 1.分配时间戳，并生成水印
        DataStream<Tuple6<Long, String, String, String, String, String>> watermarkStream = adRealTimeLogTuple
                .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness())
                //.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarksImpl());// 该构造已过时
        WatermarkStrategy
                .forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .createWatermarkGenerator()

        DataStream<Tuple6<Long, String, String, String, String, String>> filteredAdRealTimeLogStream = watermarkStream
                // 这里弃用了keyby，没必要，还浪费时间，每个key都要查一次数据库。
                .timeWindowAll(Time.seconds(3))
                .process(new FilterBlackUserProcess());
        /**
         *  统计每天每个用户对每个广告的点击量
         */

        SingleOutputStreamOperator<Tuple4<Long, Long, String, Long>> dailyUserAdClickCount = filteredAdRealTimeLogStream
                .keyBy((KeySelector<Tuple6<Long, String, String, String, String, String>, String>) value -> value.f5)
                .timeWindow(Time.seconds(3))
                .process(new DailyUserAdClickCountProcess(blacklistUserid));
        /**
         * 将dailyUserAdClickCount 流 addSink到数据库
         * 数据库端 要进行 插入？还是更新？的判别
         */
        //dailyUserAdClickCount.print();
        dailyUserAdClickCount.addSink(new DailyUserAdClickCountSink());

        /**
         * 得到 blacklistUserid 侧面输出流，并将其userid去重，然后保存到数据库
         * 继续windowALl
         */
        DataStream<Long> blacklistUseridStream = dailyUserAdClickCount.getSideOutput(blacklistUserid);

        blacklistUseridStream
                .timeWindowAll(Time.seconds(3))
                .process(new DistinctUseridProcess())
                .addSink(new BlackSink());
        //blacklistUseridStream.print();

        /**
         * 得到 filteredAdRealTimeLog 侧面输出流，以供后面的业务使用 1461756862000
         *
         * 注意： 这个侧面输出流在这得不到是因为
         *      它是从第一个window的process方法中返回的，
         *      我们却从最后一个window返回的SingleOutputStreamOperator中得到它
         *      显然是不可能的
         *      所以第一种方法是：我们要把第一个window产生的流单独得到，才能从中得到它
         *      第二种： 我们也可以从第二个窗口中返回它，第二个窗口中也是经过过滤的正常点击
         *      第三种： 也可以将两个窗口隔开，第一个窗口得到的就是经过过滤的正常点击filteredAdRealTimeLogStream
         *              这个filteredAdRealTimeLogStream后面的业务可以用到
         *              第二个窗口可以用filteredAdRealTimeLogStream开启
         *              从而得到每天每个用户对每个广告的点击量dailyUserAdClickCount
         *              再从dailyUserAdClickCount中将侧面输出流blacklistUseridStream返回
         *
         *      前面两种方法都是又将filteredAdRealTimeLogStream作为侧面输出流返回一次
         *      显然多此一举。
         *
         */

        /**
         * 业务功能一：计算广告流量实时统计结果(yyyyMMdd_province_city_adid, clickCount)
         *  最粗
         */

        // 状态dailyAdClickCount侧面输出流
        final OutputTag<Tuple2<String, Long>> dailyAdClickCountGlobal =
                new OutputTag<Tuple2<String, Long>>("dailyAdClickCountGlobal") {};

        SingleOutputStreamOperator<Tuple5<String, String, String, Long, Long>> adRealTimeStatStream = filteredAdRealTimeLogStream
                .map(new Filtered2TupleMapFunction())
                .keyBy((KeySelector<Tuple2<String, Long>, String>) value -> value.f0)
                .timeWindow(Time.seconds(3))
                .process(new DailyCityAdClickCountProcess());
        // 将窗口中的key的最新的全局count addSink到数据库，有则覆盖，无则插入
        adRealTimeStatStream.addSink(new AdRealTimeStatSink());
        //adRealTimeStatStream.print();


        /**
         * 业务功能二：实时统计每天每个省份top3热门广告
         * yyyyMMdd, province, city, adid, clickCount 已经统计了
         * 那么可以通过外部查数据库聚合出每天每个省份实时的top3热门广告
         * 所以，这一功能没必要在这统计了
         *
         *  统计的稍微细一些了
         */
        /*DataStream<Tuple2<String, Long>> dailyAdClickCountGlobalStream = adRealTimeStatStream.getSideOutput(dailyAdClickCountGlobal);
        dailyAdClickCountGlobalStream.print();*/

        /**
         * 业务功能三：实时统计每天每个广告在最近1小时的滑动窗口内的点击趋势（每分钟的点击量）
         *的非常细了
         * 我们每次都可以看到每个广告，最近一个小时内，每分钟内的点击量
         * 每支广告的点击趋势
         * watermarkStream ->  AdClickTrendSink
         * <date, hour, minute, adid, clickCount>
         */
        watermarkStream
                .map(new Stream2TimeMinuteCountMapFunction())
                .keyBy((KeySelector<Tuple2<String, Long>, String>) value -> value.f0)
                .timeWindow(Time.minutes(60), Time.seconds(10))
                .sum(1)
                .map(new TimeMinuteTuple2Tuple5())//.print();
                .addSink(new AdClickTrendSink());

        env.execute("ad");
    }

    private static FlinkKafkaConsumer011<String> getKafkaConsumer() {
        // configure Kafka consumer
        Properties props = new Properties();
        props.setProperty(Constants.ZOOKEEPER_CONNECT, ConfigurationManager.getProperty(Constants.ZOOKEEPER_CONNECT));
        props.setProperty(Constants.BOOTSTRAP_SERVERS, ConfigurationManager.getProperty(Constants.BOOTSTRAP_SERVERS));
        props.setProperty("group.id", "adRealTimeLogGroup");
        props.setProperty("auto.offset.reset", "earliest");

        // create a Kafka consumer
        return new FlinkKafkaConsumer011<>(
                        ConfigurationManager.getProperty(Constants.KAFKA_TOPICS),
                        new SimpleStringSchema(),
                        props);
    }

    private static StreamExecutionEnvironment getEnv() {
        // 构建flink流式处理程序的上下文，流式的话使用StreamExecutionEnvironment，批处理使用ExecutionEnvironment
        // 本地测试使用local environment（本地执行环境）
        StreamExecutionEnvironment env;
        if (ConfigurationManager.getBoolean(Constants.FLINK_LOCAL)) {
            env = StreamExecutionEnvironment.createLocalEnvironment();
        } else {
            env = StreamExecutionEnvironment.getExecutionEnvironment();
        }
        return env;
    }
}
